/**
 * Software License, Version 1.0 Copyright 2003 The Trustees of Indiana
 * University. All rights reserved. Redistribution and use in source and binary
 * forms, with or without modification, are permitted provided that the
 * following conditions are met: 1) All redistributions of source code must
 * retain the above copyright notice, the list of authors in the original source
 * code, this list of conditions and the disclaimer listed in this license; 2)
 * All redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the disclaimer listed in this license in the
 * documentation and/or other materials provided with the distribution; 3) Any
 * documentation included with all redistributions must include the following
 * acknowledgement: "This product includes software developed by the Community
 * Grids Lab. For further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/." Alternatively, this acknowledgement may
 * appear in the software itself, and wherever such third-party acknowledgments
 * normally appear. 4) The name Indiana University or Community Grids Lab or
 * NaradaBrokering, shall not be used to endorse or promote products derived
 * from this software without prior written permission from Indiana University.
 * For written permission, please contact the Advanced Research and Technology
 * Institute ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5)
 * Products derived from this software may not be called NaradaBrokering, nor
 * may Indiana University or Community Grids Lab or NaradaBrokering appear in
 * their name, without prior written permission of ARTI. Indiana University
 * provides no reassurances that the source code provided does not infringe the
 * patent or any other intellectual property rights of any other entity. Indiana
 * University disclaims any liability to any recipient for claims brought by any
 * other entity based on infringement of intellectual property rights or
 * otherwise. LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH
 * NO WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY
 * GIVES NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
/**
 * Software License, Version 1.0 Copyright 2003 The Trustees of Indiana
 * University. All rights reserved. Redistribution and use in source and binary
 * forms, with or without modification, are permitted provided that the
 * following conditions are met: 1) All redistributions of source code must
 * retain the above copyright notice, the list of authors in the original source
 * code, this list of conditions and the disclaimer listed in this license; 2)
 * All redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the disclaimer listed in this license in the
 * documentation and/or other materials provided with the distribution; 3) Any
 * documentation included with all redistributions must include the following
 * acknowledgement: "This product includes software developed by the Community
 * Grids Lab. For further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/." Alternatively, this acknowledgement may
 * appear in the software itself, and wherever such third-party acknowledgments
 * normally appear. 4) The name Indiana University or Community Grids Lab or
 * NaradaBrokering, shall not be used to endorse or promote products derived
 * from this software without prior written permission from Indiana University.
 * For written permission, please contact the Advanced Research and Technology
 * Institute ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5)
 * Products derived from this software may not be called NaradaBrokering, nor
 * may Indiana University or Community Grids Lab or NaradaBrokering appear in
 * their name, without prior written permission of ARTI. Indiana University
 * provides no reassurances that the source code provided does not infringe the
 * patent or any other intellectual property rights of any other entity. Indiana
 * University disclaims any liability to any recipient for claims brought by any
 * other entity based on infringement of intellectual property rights or
 * otherwise. LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH
 * NO WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY
 * GIVES NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.discovery.broker;

import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.HttpURLConnection;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.URL;
import java.net.UnknownHostException;
import java.util.Hashtable;
import java.util.Properties;
import java.util.Random;
import java.util.StringTokenizer;
import java.util.Vector;

import org.apache.log4j.Logger;

import cgl.narada.discovery.broker.messages.BrokerAdvertisement;
import cgl.narada.discovery.broker.messages.BrokerDiscoveryRequest;
import cgl.narada.discovery.broker.messages.BrokerDiscoveryResponse;
import cgl.narada.protocol.Protocol;
import cgl.narada.service.ServiceException;
import cgl.narada.service.ServiceProperties;
import cgl.narada.service.ServicePropertiesFactory;
import cgl.narada.service.timer.ClockI;
import cgl.narada.service.timer.HRClock;
import cgl.narada.transport.util.BytesUtil;
import cgl.narada.util.SystemInit;
import cgl.narada.util.SystemNotInitializedException;

/**
 * A client can use the Broker Locator to locate a suitable broker to connect
 * to. <br>
 * The BrokerDiscoveryHelper sends <code>BrokerDiscoveryRequest</code> to (a)
 * known (BDN)s. If the BDN is not available then the request is broadcasted
 * using UDP multicast. <br>
 * Created on Mar 26, 2005
 * 
 * @author Harshawardhan Gadgil (hgadgil@grids.ucs.indiana.edu)
 */

// NOTE: Remove Code marked with BENCHMARK: for distribution version
// ====
public class BrokerDiscoveryHelper {

    static Logger log = Logger.getLogger("BrokerDiscoveryHelper");

    // Datagram socket timeout in milliseconds
    public static int TIMEOUT = 5000;

    // Big enuff buffer for BrokerDiscoveryResponse Packet (Bytes)
    // TODO: Whats the MAX size this can go to ? OR Configure from command line
    // /
    // config file !!
    public static final int RESPONSE_PACKET_SIZE = 4096;

    // Max number of responses to wait for
    private int MAX_RESPONSES = 3;

    // We ping only the BROKER_SET_SIZE number of brokers
    // NOTE: BROKER_SET_SIZE <= MAX_RESPONSES
    private int BROKER_TARGET_SET_SIZE = 3;

    private String[] BDN;

    private DatagramSocket udpSocket;

    // If the request times out, we simply wait for first response for an
    // additional TIMEOUT
    // interval.
    private boolean requestTimedOut;

    // tracks the number of responses recieved
    private int numResponses;

    private Vector discoveryResponses;

    private DatagramPacket responsePacket;

    private static String op; // testing purposes only.

    // How should the discovery response metrics be weighed ?
    // This should be configured from the config file... and would determine how
    // the "BEST BROKER" to connect to is selected.

    /** 50% weightage is given to the Ping time */
    public static final double BROKER_DISCOVERY_PING_TIME = 0.9;

    /** 10% weightage is given to the amount of memory present */
    public static final double WEIGHTAGE_TOTAL_MEMORY = 0.01;

    /** 5% weightage is given to the Number of links a broker has */
    public static final double WEIGHTAGE_NUM_LINKS = 0.05;

    /** 4% weightage is given to the %age of free memory */
    public static final double WEIGHTAGE_FREE_TO_TOTAL_MEMORY = 0.04;

    // Some other constants...
    public static final int MAX_PINGS = 1;

    // Some objects frequently used but not required to be initialized for each
    // use
    private Random pingIDGenerator = new Random(65535);

    private ClockI hrClock;

    private ServiceProperties serviceProperties;

    private boolean benchmark = false;


    public BrokerDiscoveryHelper(int entityId, String configFilePath,
            int timeout, int maxResponses, int maxSetSize) {

        ServicePropertiesFactory servicePropertiesFactory =
            ServicePropertiesFactory.getInstance();
        try {
            servicePropertiesFactory.intializeServiceProperties(entityId,
                                                                configFilePath);
        } catch (ServiceException e1) {
            log.error("", e1);
            return;
        }
        serviceProperties =
            servicePropertiesFactory.getServiceProperties(entityId);

        if (benchmark) {
            // Create the clock
            String nbHome = null;
            try {
                nbHome = SystemInit.getInitData(SystemInit.SYS_NB_HOME);
            } catch (SystemNotInitializedException e) {
                log.error("", e);
            }

            hrClock = HRClock.getClock(nbHome);
        }

        // At least 1 response
        if (maxResponses < 1)
            maxResponses = 1;

        TIMEOUT = timeout;
        MAX_RESPONSES = maxResponses;
        BROKER_TARGET_SET_SIZE =
            (maxSetSize > MAX_RESPONSES) ? MAX_RESPONSES : maxSetSize;

        // Load a list of BDNs

        String bdnList = "";
        if (serviceProperties.hasProperty("BDNDiscoveryList")) {
            bdnList = serviceProperties.getProperty("BDNDiscoveryList");
        }

        if (bdnList.startsWith("[")) {
            bdnList = bdnList.substring(1);
        }

        if (bdnList.endsWith("]")) {
            bdnList = bdnList.substring(0, bdnList.length() - 1);
        }
        StringTokenizer st = new StringTokenizer(bdnList, ",");

        BDN = new String[st.countTokens()];
        int i = 0;
        while (st.hasMoreTokens()) {
            BDN[i++] = st.nextToken();
        }

        discoveryResponses = new Vector();

        try {
            udpSocket = new DatagramSocket();

        } catch (SocketException e) {
            log.error("", e);
        }

        responsePacket =
            new DatagramPacket(new byte[RESPONSE_PACKET_SIZE],
                               RESPONSE_PACKET_SIZE);
    }


    /**
     * Sends the Discovery request to available BDN's. If none of the listed
     * BDNs are contactable, the request is sent using multicast
     * 
     * @param protocols -
     *            An array of strings representing the protocols the broker must
     *            support
     * @param securityCred -
     *            Certificate, TODO: Replace w/ X.509 Certificate
     * @param matchingVirtualBrokerNetworks -
     *            Brokers belongign to which networks must be considered
     * @return True if the request could be sent, false otherwise
     */
    private boolean sendDiscoveryRequest(String[] protocols,
                                         String securityCred,
                                         String matchingVirtualBrokerNetworks) {
        long start = 0, end = 0;

        // BENCHMARK:
        if (benchmark) {
            start = hrClock.getTimeMicroseconds();
        }

        BrokerDiscoveryRequest discoRequest = new BrokerDiscoveryRequest();
        discoRequest.setPort(udpSocket.getLocalPort());
        discoRequest.setTransportProtocols(protocols);
        discoRequest.setSecurityCredentials(securityCred); // Unused for now...
        discoRequest.setBelongingToNetwork(matchingVirtualBrokerNetworks);

        byte[] discoBytes = discoRequest.getBytes();

        // BENCHMARK
        if (benchmark) {
            end = hrClock.getTimeMicroseconds();
            System.out
                .println("BENCHMARK# BrokerDiscoRequest Construction (microsec): "
                    + (end - start));
        }

        // Send it to the BDNs, Currently we just send it to 1 BDN. The process
        // stops after first successful send.

        boolean success = false;
        int i = 0;

        // BENCHMARK
        if (benchmark) {
            start = hrClock.getTimeMicroseconds();
        }

        while ((!success) && (i < BDN.length)) {
            System.out.println("Trying: " + BDN[i]);

            try {
                URL url = new URL(BDN[i]);
                HttpURLConnection urlConnection =
                    (HttpURLConnection) url.openConnection();

                urlConnection.setRequestMethod("POST");
                urlConnection.setDoInput(true);
                urlConnection.setDoOutput(true);
                urlConnection.setUseCaches(false);

                urlConnection.setRequestProperty("CONTENT_LENGTH", ""
                    + discoBytes.length);
                OutputStream os = urlConnection.getOutputStream();
                os.write(discoBytes);
                os.flush();
                os.close();

                // Now read the Response associated with this request
                BufferedReader br =
                    new BufferedReader(new InputStreamReader(urlConnection
                        .getInputStream()));
                String line;
                while ((line = br.readLine()) != null) {
                    log.info("BDN<" + BDN[i] + "> Sends: " + line);
                    // May want to change these semantics in future
                    if (line.startsWith("OK")) {
                        success = true;
                        break;
                    }

                    if (line.startsWith("ERROR")) {
                        success = false;
                        break;
                    }

                }
            } catch (Exception e) {
                log.warn("Unable to publish <" + BDN[i] + "> :" + e);
            }

            ++i;
        }

        // BENCHMARK
        if (benchmark) {
            end = hrClock.getTimeMicroseconds();
            System.out.println("BENCHMARK# FWD Request to " + BDN.length
                + " BDN(s) (microsec): " + (end - start));
        }

        // Also multicast the discovery request. If one or more broker are not
        // registered in the BDN node then they would get the request from
        // multicast. Duplicate requests are automatically dropped.
        // The multicast address is configured in BrokerConfiguration.txt

        // NOTE:
        // 1. A limitation is that the multicast traffic must be enabled for
        // remote brokers.

        // 2. Also the broker config file on client's side may specify a
        // specific
        // multicast port, however brokers configured for different multicast
        // ports would have to rely on either the BDN sending them the request
        // OR
        // 'a broker connected to' is forwarding the request

        if (!serviceProperties.hasProperty("MulticastGroupHost")
            && !serviceProperties.hasProperty("MulticastGroupPort")) {
            return success;
        }

        String multicastAddr =
            serviceProperties.getProperty("MulticastGroupHost");
        int multicastPort =
            Integer.parseInt(serviceProperties
                .getProperty("MulticastGroupPort"));
        if (multicastPort > 0) {
            log.info("Trying multicasting request to " + multicastAddr + ":"
                + multicastPort);

            // BENCHMARK
            if (benchmark) {
                start = hrClock.getTimeMicroseconds();
            }
            try {
                DatagramSocket socket = new DatagramSocket();
                InetAddress multicastGroup =
                    InetAddress.getByName(multicastAddr);

                // Protocol.DATA + Protocol.DISCOVERY_REQUEST + request
                // byte[] packet = new byte[1 + 1 + discoBytes.length];

                ByteArrayOutputStream bout = new ByteArrayOutputStream();
                bout.write(new byte[] { Protocol.DATA });
                bout.write(new byte[] { Protocol.DISCOVERY_REQUEST });
                bout.write(discoBytes);

                byte[] bytePacket = bout.toByteArray();

                DatagramPacket packet =
                    new DatagramPacket(bytePacket, bytePacket.length,
                                       multicastGroup, multicastPort);
                socket.send(packet);
                success = true;
		socket.close();
            } catch (SocketException e) {
                log.error("", e);
            } catch (UnknownHostException e) {
                log.error("", e);
            } catch (IOException e) {
                log.error("", e);
	    }
	    
        } else {
            log.warn("Multicast Port not defined in Service Config !");
        }

        // BENCHMARK
        if (benchmark) {
            end = hrClock.getTimeMicroseconds();
            System.out.println("BENCHMARK# Multicast DiscoRequest(microsec): "
                + (end - start));
        }

        return success;
    }


    /**
     * We first initialize and run the UDPReciever thread. The try to send the
     * request to one of the BDNs of if unsuccessful, try and multicast the
     * request. In either case, if the request could be sent, we wait for
     * responses else we close the UDPReciever's socket and return null.
     * 
     * @param protocols -
     *            A list of 1 or more matching protocols the broker MUST
     *            support.
     * @param secCerts -
     *            Security Certificate. TODO: Replace w/ X.509 Certificate
     * @param brokerNetworks -
     *            brokers belonging to which networks must be considered. "*" to
     *            match any network
     * @return null if a BrokerDiscoveryRequest could not be sent, an array of
     *         Responses (possibly empty) otherwise.
     */
    public BrokerDiscoveryResponse[] discover(String[] protocols,
                                              String secCerts,
                                              String brokerNetworks) {

        long start = 0, end = 0;

        discoveryResponses.clear(); // Empty previous results if any

        try {
            udpSocket.setSoTimeout(TIMEOUT);
        } catch (SocketException e1) {
            log.error("", e1);
        }

        if (sendDiscoveryRequest(protocols, secCerts, brokerNetworks)) {
            numResponses = 0;

            // BENCHMARK
            if (benchmark) {
                start = hrClock.getTimeMicroseconds();
            }

            while (numResponses < MAX_RESPONSES) {
                try {
                    udpSocket.receive(responsePacket);
                } catch (Exception e) {
                    if (e instanceof SocketTimeoutException) {
                        // Give up ?? no broker present OR nobody wants to talk
                        // to me OR some have already responded !!
                        log.info("TIMEOUT waiting for responses");
                        break;
                    } else {
                        if (benchmark) {
                            log.error("", e);
                        }
                    }
                    // Continue until atleast max_responses have been received
                    // OR a
                    // timeout occurs.
                    continue;
                }

                // IF we come here, then the udpSocket.receive was successful
                log.info("Discovery Response: [" + responsePacket.getAddress()
                    + ":" + responsePacket.getPort() + "]");
                numResponses++;
                BrokerDiscoveryResponse bdresp =
                    BrokerDiscoveryResponse
                        .createObjectFromBytes(responsePacket.getData());
                discoveryResponses.add(bdresp);
            }

            // BENCHMARK
            if (benchmark) {
                end = hrClock.getTimeMicroseconds();
                System.out.println("BENCHMARK# Gathering "
                    + discoveryResponses.size() + " responses(microsec): "
                    + (end - start));
            }

            return (BrokerDiscoveryResponse[]) discoveryResponses
                .toArray(new BrokerDiscoveryResponse[discoveryResponses.size()]);

        } else
            return null;
    }


    /**
     * Pings a given broker
     * 
     * @param host -
     *            IP address of the broker
     * @param port -
     *            UDP port of broker
     * @param timeout -
     *            Time for which to wait
     * @return PING time in microSeconds if successful, -1 if timeout, -2 if
     *         garbled protocol header or other error.
     */
    public long ping(String host, int port, int timeout) {

        try {
            DatagramSocket socket = new DatagramSocket();

            // Quick FIX IFF previous ping time was < 1 mSec
            if (timeout <= 0)
                timeout = 1;

            socket.setSoTimeout(timeout);

            byte[] data = new byte[9];
            long pingId = pingIDGenerator.nextLong();

            data[0] = Protocol.PING;
            byte[] idBytes = BytesUtil.longToBytes(pingId);
            System.arraycopy(idBytes, 0, data, 1, 8);

            InetAddress addr = InetAddress.getByName(host);

            DatagramPacket packet =
                new DatagramPacket(data, data.length, addr, port);

            long start = hrClock.getTimeMicroseconds();

            socket.send(packet);

            while (true) {
                socket.receive(packet);
                if (packet.getData()[0] == Protocol.PING_REPLY) {
                    System.arraycopy(packet.getData(), 1, idBytes, 0, 8);
                    long rcvdId = BytesUtil.bytesToLong(idBytes);
                    if (rcvdId == pingId) {
                        break;
                    } else {
                        // probably a previous PING has reached now ! so discard
                        // it
                        log
                            .warn("** Ping_REPLY with wrong ID recieved ! Discarded !!");
                    }
                }
            }

            if (packet.getData()[0] == Protocol.PING_REPLY) {
                long end = hrClock.getTimeMicroseconds();
                return end - start;
            } else {
                log.info("Garbled Protocol Header recieved !");
                return -2;
            }
        } catch (Exception e) {
            if (e instanceof SocketTimeoutException) {
                log.info("TIMEOUT waiting for PING response from {" + host
                    + ":" + port + "}");
                return -1;
            }
            log.error("", e);
            return -2;
        }

    }


    /**
     * Computes weight for the given broker using the values in the
     * BrokerDiscoveryResponse
     * 
     * @param resp -
     *            The BrokerDiscoveryResponse containing the metrics for the
     *            broker
     * @return - A double value indicating the computed weight
     */
    public double computeBrokerWeightage(BrokerDiscoveryResponse resp) {
        // TODO: Get some metric of CPU load somehow
        // double cpu = brokers[i].GET_CPU();
        double freeMem = resp.getFreeMemory();
        double totalMem = resp.getTotalmemory();
        double numLinks = resp.getNumLinks();
        double weight = 0;

        // + cpu ??

        // Higher the better
        weight += (freeMem / totalMem) * WEIGHTAGE_FREE_TO_TOTAL_MEMORY;
        weight += (totalMem / (1024 * 1024)) * WEIGHTAGE_TOTAL_MEMORY;

        // Lower the better
        weight -= numLinks * WEIGHTAGE_NUM_LINKS;

        // weight = (double) Math.round(weight * 1000.0) / 1000.0;
        // log.info("Broker_Wt.[" + getBrokerLocation(resp) + "]: " + weight);
        return weight;
    }


    public String getBrokerLocation(BrokerDiscoveryResponse bdr) {
        String broker = bdr.getBrokerInfo().getInfo();
        return broker.substring(broker.indexOf("#") + 1);
    }


    /**
     * Selects the best broker among the specified set of brokers from the
     * broker discovery responses
     * 
     * @param brokers
     *            An array of BrokerDiscoveryResponses
     * @return The BrokerDiscoveryResponse of the best broker
     */
    public BrokerDiscoveryResponse selectBestBroker(
                                                    BrokerDiscoveryResponse[] brokers) {

        op = "";

        // BENCHMARK
        long start = 0, end = 0;

        if (benchmark) {
            start = hrClock.getTimeMicroseconds();
        }

        // Step 1: compute the weights for responding brokers
        double[] weights = new double[brokers.length];

        for (int i = 0; i < brokers.length; i++) {
            weights[i] = computeBrokerWeightage(brokers[i]);
        }

        // BENCHMARK
        if (benchmark) {
            end = hrClock.getTimeMicroseconds();
            System.out.println("BENCHMARK# Computing weights for "
                + brokers.length + " responses(microsec): " + (end - start));
        }

        // BENCHMARK
        if (benchmark) {
            start = hrClock.getTimeMicroseconds();
        }
        // Step 2: Now generate the BROKER_TARGET_SET depending on the weights
        // of
        // responding brokers
        BrokerDiscoveryResponse[] BROKER_TARGET_SET;

        if (brokers.length <= BROKER_TARGET_SET_SIZE) {
            // No need to compute the target set, since we'll use all entries
            BROKER_TARGET_SET = new BrokerDiscoveryResponse[brokers.length];

        } else {
            // Since we have more brokers we need to short-list the target set
            BROKER_TARGET_SET =
                new BrokerDiscoveryResponse[BROKER_TARGET_SET_SIZE];

            // Step 2a: A simple bubble sort on the weights.

            for (int i = 0; i < brokers.length; i++) {
                for (int j = i + 1; j < brokers.length; j++) {
                    if (weights[i] < weights[j]) {
                        double tmp = weights[j]; // swap a[j] and a[j+1]
                        weights[j] = weights[i];
                        weights[i] = tmp;

                        // Also swap the corresponding brokers[] entries
                        BrokerDiscoveryResponse tmpResp = brokers[j];
                        brokers[j] = brokers[i];
                        brokers[i] = tmpResp;
                    }
                }
            }
        }

        // Step 3: Select the first BROKER_TARGET_SET_SIZE brokers
        System.arraycopy(brokers, 0, BROKER_TARGET_SET, 0,
                         BROKER_TARGET_SET.length);

        // BENCHMARK
        if (benchmark) {
            end = hrClock.getTimeMicroseconds();
            System.out.println("BENCHMARK# Generating Broker target set "
                + BROKER_TARGET_SET_SIZE + "/" + brokers.length
                + " (microsec): " + (end - start));
        }

        // BENCHMARK
        if (benchmark) {
            start = hrClock.getTimeMicroseconds();
        }

        // Step 4: For each broker in the BROKER_TARGET_SET, determine the
        // nearest
        // broker by pinging it MAX_PINGS times and compute the average ping
        // time

        long lowestPingTime = 5000;
        int selectedIndex = 0;

        for (int i = 0; i < BROKER_TARGET_SET.length; i++) {

            BrokerDiscoveryResponse resp = BROKER_TARGET_SET[i];
            BrokerAdvertisement bAd = resp.getBrokerInfo();
            String host = bAd.getHostname();
            int port = 0;

            Hashtable protocols = bAd.getSupportedTransportProtocols();
            Properties udpProps = (Properties) protocols.get("udp");
            port = Integer.parseInt(udpProps.getProperty("UDPBrokerPort"));

            if (port == 0) {
                log.warn("Unable to determine UDP port for broker: " + host);
                continue;
            }

            // Ping MAX_PINGS times to find average ping time
            long pingTime = 0;

            for (int pingCounter = 0; pingCounter < MAX_PINGS; pingCounter++) {
                System.out.println("TimeOUT: " + lowestPingTime);
                long p = ping(host, port, (int) lowestPingTime);

                // NOTE: This is an arbitrary value signifying that the broker
                // was
                // unreachable
                if (p < 0)
                    p = 10000;
                pingTime += p;
            }

            pingTime /= MAX_PINGS;

            // Convert to millisecs !
            pingTime = Math.round(pingTime / 1000);

            // System.out.println("METRICS FOR: " +
            // brokers[i].getBrokerInfo().getInfo());

            System.out.println("PING[" + getBrokerLocation(resp) + "]: "
                + pingTime + " mSec");

            // This is for benchmarking purposes only =====
            String selBrok = resp.getBrokerInfo().getInfo();
            int selBrokId = 0;

            if (selBrok.endsWith("COMPLEXITY.UCS.INDIANA.EDU"))
                selBrokId = 2;
            else if (selBrok.endsWith("BOUSCAT.CS.CF.AC.UK"))
                selBrokId = 3;
            else if (selBrok.endsWith("PAMD2.CSIT.FSU.EDU"))
                selBrokId = 4;
            else if (selBrok.endsWith("WEBIS.MSI.UMN.EDU"))
                selBrokId = 5;
            else if (selBrok.endsWith("TUNGSTEN.NCSA.UIUC.EDU"))
                selBrokId = 6;

            op += selBrokId + " " + pingTime + " ";
            // =======

            // System.out.println("\t NUM LINKS: " + brokers[i].getNumLinks());
            // System.out.println("\t USED MEM : " + usedMem + " MB");
            // System.out.println("\t WEIGHT : " + weight + "\t\tLW: " +
            // lowest_weight);

            if (pingTime < lowestPingTime) {
                lowestPingTime = pingTime;
                selectedIndex = i;
            }
            // System.out.println("BrokerDiscoveryHelper.selectBestBroker():
            // SELECTED_INDEX:"+ selectedIndex);
        }

        // BENCHMARK
        if (benchmark) {
            end = hrClock.getTimeMicroseconds();
            System.out.println("BENCHMARK# Deciding Best Broker(microsec): "
                + (end - start));
        }

        // OK all weights gathered !!
        return BROKER_TARGET_SET[selectedIndex];
    }


    /**
     * Testing purposes only
     * 
     * @param args
     */
    public static void main(String[] args) {

        if (args.length != 6) {

            System.out.println("ARGS.LEN: " + args.length);
            for (int i = 0; i < args.length; i++) {
                System.out.println("ARG[" + i + "]: " + args[i]);
            }

            System.out.println("USAGE: ");
            System.out
                .println("java -classpath CP cgl.narada.discovery.broker.BrokerDiscoveryHelper <<params>>");
            System.out.println("<<PARAMS>> are as follows");
            System.out
                .println("<configFileName> <timeOut> <maxResponses> <targetSetSize> <maxIter> <outputFile>");
            System.exit(-1);
        }

        // MUST do this...
        SystemInit.init();

        // HOW TO USE ??
        // -------

        int timeout = Integer.parseInt(args[1]);
        int maxResp = Integer.parseInt(args[2]);
        int targetSetSize = Integer.parseInt(args[3]);
        int maxIter = Integer.parseInt(args[4]);
        String outputTiming = args[5];

        BrokerDiscoveryHelper bloc =
            new BrokerDiscoveryHelper(2000, args[0], timeout, maxResp,
                                      targetSetSize);
        BrokerDiscoveryResponse[] resp;

        String[] protocolsSet_1 = { "tcp", "niotcp" };
        resp = bloc.discover(protocolsSet_1, "*SECURITY_CERT*", "*");
        System.out.println("Total Responses : " + resp.length);
        for (int i = 0; i < resp.length; i++) {
            System.out.println("Found BROKER @: "
                + resp[i].brokerInfo.getHostname());
            System.out.println("======\n" + resp[i].toString() + "\n=======");
        }

        // ------------

        // System.out.println("\n\n====TEST 1 (BAD Protocol)====== \n");
        // String[] protocolsSet_1 = { "tcp", "niotcp", "udp", "YeeHAW" };
        // resp = bloc.discover(protocolsSet_1, "*SECURITY_CERT*", "*");
        // System.out.println("Total Responses : " + resp.length);
        // for (int i = 0; i < resp.length; i++) {
        // System.out.println("Found BROKER @: "
        // + resp[i].brokerInfo.getHostname());
        // System.out.println("======\n" + resp[i].toString() + "\n=======");
        // }
        //
        // System.out.println("\n\n====TEST 2 (* <- Network)====== \n");
        //
        // String[] protocolsSet_2 = { "tcp", "niotcp" };
        // resp = bloc.discover(protocolsSet_2, "*SECURITY_CERT*", "*");
        // System.out.println("Total Responses : " + resp.length);
        // for (int i = 0; i < resp.length; i++) {
        // System.out.println("Found BROKER @: "
        // + resp[i].brokerInfo.getHostname());
        // System.out.println("======\n" + resp[i].toString() + "\n=======");
        // }
        //
        // System.out
        // .println("\n\n=====TEST 3 (Random[non-existing] Network)===== \n");
        //
        // String[] protocolsSet_3 = { "tcp", "niotcp" };
        // resp = bloc.discover(protocolsSet_3, "*SECURITY_CERT*", "CGL-2");
        // System.out.println("Total Responses : " + resp.length);
        // for (int i = 0; i < resp.length; i++) {
        // System.out.println("Found BROKER @: "
        // + resp[i].brokerInfo.getHostname());
        // System.out.println("======\n" + resp[i].toString() + "\n=======");
        // }

        // System.out.println("\n\n=====TEST 4 (exiting network)===== \n");

        // String[] protocolsSet_4 = { "tcp", "niotcp" };
        // FileOutputStream fout = null;
        // try {
        // fout = new FileOutputStream(outputTiming);
        // } catch (FileNotFoundException e) {
        // // TODO Auto-generated catch block
        // // e.printStackTrace();
        // log.error("", e);
        // }
        // ClockI hrclock = HRClock.getClock();
        //
        // Object o = new Object();
        // for (int i = 0; i < maxIter; i++) {
        //
        // long start = hrclock.getTimeMicroseconds();
        //
        // resp = bloc.discover(protocolsSet_4, "*SECURITY_CERT*",
        // "network-CGL-1");
        // // System.out.println("Total Responses : " + resp.length);
        // // for (int i = 0; i < resp.length; i++) {
        // // System.out.println("Found BROKER @: "
        // // + resp[i].brokerInfo.getHostname());
        // // System.out.println("======\n" + resp[i].toString() + "\n=======");
        // // }
        //
        // BrokerDiscoveryResponse selectedBroker = bloc.selectBestBroker(resp);
        // long end = hrclock.getTimeMicroseconds();
        //
        // String selBrok = selectedBroker.getBrokerInfo().getInfo();
        // int selBrokId = 0;
        //
        // if (selBrok.endsWith("COMPLEXITY.UCS.INDIANA.EDU"))
        // selBrokId = 2;
        // else if (selBrok.endsWith("BOUSCAT.CS.CF.AC.UK"))
        // selBrokId = 3;
        // else if (selBrok.endsWith("PAMD2.CSIT.FSU.EDU"))
        // selBrokId = 4;
        // else if (selBrok.endsWith("WEBIS.MSI.UMN.EDU"))
        // selBrokId = 5;
        // else if (selBrok.endsWith("TUNGSTEN.NCSA.UIUC.EDU")) selBrokId = 6;
        //
        // try {
        // fout.write(("" + (end - start) + " " + selBrokId + "\t"
        // + BrokerDiscoveryHelper.op + "\r\n").getBytes());
        // fout.flush();
        // } catch (IOException e1) {
        // // TODO Auto-generated catch block
        // // e1.printStackTrace();
        // log.error("", e1);
        // }
        //
        // System.out.println("======BEST BROKER: "
        // + selectedBroker.getBrokerInfo().getInfo());
        // try {
        // // Wait for systems to catch breath :)
        // synchronized (o) {
        // o.wait(1000);
        // }
        // } catch (InterruptedException e2) {
        // // TODO Auto-generated catch block
        // // e2.printStackTrace();
        // log.error("", e2);
        // }
        // }
        // System.out.println("\n\n======BEST BROKER:=====\n"
        // + selectedBroker.toString() + "\n=========================");
    }

}
